/*
 *   Copyright 2021 zzh
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.aduib.boot.common.concurrent;

import com.aduib.boot.common.log.LOG;
import com.aduib.boot.common.util.SystemUtils;
import org.jctools.maps.NonBlockingHashMap;
import org.jctools.maps.NonBlockingHashMapLong;
import org.jctools.maps.NonBlockingHashSet;
import org.jctools.queues.MpmcArrayQueue;
import org.jctools.queues.MpscArrayQueue;
import org.jctools.queues.MpscChunkedArrayQueue;
import org.jctools.queues.MpscUnboundedArrayQueue;
import org.jctools.queues.SpmcArrayQueue;
import org.jctools.queues.SpscArrayQueue;
import org.jctools.queues.SpscLinkedQueue;
import org.jctools.queues.SpscUnboundedArrayQueue;
import org.jctools.queues.atomic.MpscAtomicArrayQueue;
import org.jctools.queues.atomic.MpscChunkedAtomicArrayQueue;
import org.jctools.queues.atomic.MpscUnboundedAtomicArrayQueue;
import org.jctools.queues.atomic.SpscAtomicArrayQueue;
import org.jctools.queues.atomic.SpscLinkedAtomicQueue;
import org.jctools.util.Pow2;
import org.jctools.util.UnsafeAccess;

import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Queue;

import static java.lang.Math.max;
import static java.lang.Math.min;

/**
 * @description: Jctools相关类相关类快捷创建
 * @author: zzh
 * @date: 2021/9/6 14:05
 */
public class JctoolsHelper {

  private static final int MPSC_CHUNK_SIZE =  1024;

  private static final int MIN_MAX_MPSC_CAPACITY =  MPSC_CHUNK_SIZE * 2;

  private static final int MAX_ALLOWED_MPSC_CAPACITY = Pow2.MAX_POW2;

  /**
   * 对 ConcurrentHashMap 的增强，对多 CPU 的支持以及高并发更新提供更好的性能。
   *
   * @return org.jctools.maps.NonBlockingHashMap
   */
  public static <K, T> NonBlockingHashMap<K, T> nonBlockingHashMap() {
    return new NonBlockingHashMap<>();
  }

  public static <K, T> NonBlockingHashMap<K, T> nonBlockingHashMap(int initialCapacity) {
    return new NonBlockingHashMap<>(initialCapacity * 16);
  }

  public static <E> NonBlockingHashSet<E> nonBlockingHashSet() {
    return new NonBlockingHashSet<>();
  }

  /**
   * key 为 Long 型的 NonBlockingHashMap。
   *
   * @param <T>
   * @return org.jctools.maps.NonBlockingHashMapLong
   */
  public static <T> NonBlockingHashMapLong<T> nonBlockingHashMapLong() {
    return new NonBlockingHashMapLong<>();
  }

  public static <T> NonBlockingHashMapLong<T> nonBlockingHashMapLong(int initialCapacity) {
    return new NonBlockingHashMapLong<>(initialCapacity * 16);
  }

  /**
   * 单一生产者单一消费-有界(初始容量建议是4的倍数)
   *
   * @param capacity
   * @param <T>
   * @return org.jctools.queues.SpscArrayQueue
   */
  public static <T> Queue<T> spscArrayQueue(int capacity) {
    return SystemUtils.hasUnsafe()?new SpscArrayQueue<T>(capacity):new SpscAtomicArrayQueue<T>(capacity);
  }

  /**
   * 单一生产者单一消费者-无界(初始容量建议是16的倍数)
   *
   * @param initialCapacity
   * @param <T>
   * @return
   */
  public static <T> SpscUnboundedArrayQueue<T> spscUnboundedArrayQueue(int initialCapacity) {
    return new SpscUnboundedArrayQueue<>(initialCapacity * 16);
  }

  /**
   * 单一生产者单一消费者-无界
   * @param <T>
   * @return
   */
  public static <T> Queue<T> spscUnboundedQueue() {
    return SystemUtils.hasUnsafe()?new SpscLinkedQueue<T>():new SpscLinkedAtomicQueue<T>();
  }

  /**
   * 单生产者多消费者（有界）
   *
   * @param initialCapacity
   * @param <T>
   * @return
   */
  public static <T> SpmcArrayQueue<T> spmcArrayQueue(int initialCapacity) {
    return new SpmcArrayQueue<>(initialCapacity * 16);
  }

  /**
   * 多生产者多消费者-无界(初始容量建议是2的倍数)
   *
   * @param initialCapacity
   * @param <T>
   * @return
   */
  public static <T> MpmcArrayQueue<T> mpmcArrayQueue(int initialCapacity) {
    return new MpmcArrayQueue<>(initialCapacity * 2);
  }

  /**
   * 多生产者单一消费者（有界和无界）
   *
   * @param fixCapacity
   * @param <T>
   * @return
   */
  public static <T> Queue<T> mpscArrayQueue(int fixCapacity) {
    return SystemUtils.hasUnsafe()?new MpscArrayQueue<T>(fixCapacity):new MpscAtomicArrayQueue<T>(fixCapacity);
  }

  /**
   * 多生产者单一消费者（有界和无界）
   *
   * @param <T>
   * @return
   */
  public static <T> Queue<T> mpscArrayQueue() {
    return Mpsc.newMpscQueue();
  }

  public static <T> Queue<T> mpscArrayQueue2(final int maxCapacity) {
    return Mpsc.newMpscQueue(maxCapacity);
  }

  /**
   * 多生产者单一消费者（有界和无界）
   *
   * @param maxCapacity
   * @param <T>
   * @return
   */
  @Deprecated
  public static <T> MpscChunkedArrayQueue<T> mpscChunkedArrayQueue(int maxCapacity) {
    return new MpscChunkedArrayQueue<>(1024, maxCapacity * 1024);
  }
  
  /**
   * 多生产者单一消费者（有界和无界）
   *
   * @param initialCapacity
   * @param <T>
   * @return
   */
  @Deprecated
  public static <T> MpscUnboundedArrayQueue<T> mpscUnboundedArrayQueue(int initialCapacity) {
    return new MpscUnboundedArrayQueue<>(initialCapacity * 16);
  }

  private static final class Mpsc {
    private static final boolean USE_MPSC_CHUNKED_ARRAY_QUEUE;

    static {
      Object unsafe = null;
      if (SystemUtils.hasUnsafe()) {
        // jctools goes through its own process of initializing unsafe; of
        // course, this requires permissions which might not be granted to calling code, so we
        // must mark this block as privileged too
        unsafe = AccessController.doPrivileged(new PrivilegedAction<Object>() {
          @Override
          public Object run() {
            // force JCTools to initialize unsafe
            return UnsafeAccess.UNSAFE;
          }
        });
      }

      if (unsafe == null) {
        LOG.d("org.jctools-core.MpscChunkedArrayQueue: unavailable");
        USE_MPSC_CHUNKED_ARRAY_QUEUE = false;
      } else {
        LOG.d("org.jctools-core.MpscChunkedArrayQueue: available");
        USE_MPSC_CHUNKED_ARRAY_QUEUE = true;
      }
    }

    private Mpsc() {
    }

    static <T> Queue<T> newMpscQueue(final int maxCapacity) {
      // Calculate the max capacity which can not be bigger than MAX_ALLOWED_MPSC_CAPACITY.
      // This is forced by the MpscChunkedArrayQueue implementation as will try to round it
      // up to the next power of two and so will overflow otherwise.
      final int capacity = max(min(maxCapacity, MAX_ALLOWED_MPSC_CAPACITY), MIN_MAX_MPSC_CAPACITY);
      return USE_MPSC_CHUNKED_ARRAY_QUEUE ? new MpscChunkedArrayQueue<T>(MPSC_CHUNK_SIZE, capacity)
              : new MpscChunkedAtomicArrayQueue<T>(MPSC_CHUNK_SIZE, capacity);
    }

    static <T> Queue<T> newMpscQueue() {
      return USE_MPSC_CHUNKED_ARRAY_QUEUE ? new MpscUnboundedArrayQueue<T>(MPSC_CHUNK_SIZE)
              : new MpscUnboundedAtomicArrayQueue<T>(MPSC_CHUNK_SIZE);
    }
  }
}
